package org.opens.lambda.parallel;

import com.google.common.base.Stopwatch;
import org.junit.Test;

import java.util.stream.LongStream;
import java.util.stream.Stream;

public class ParallelProcessing {

    /**
     * 功能:
     *      使用java的api进行cpu核心数量检查.
     * 输出:
     *      4
     * 说明:
     *      1. 输出的数量表示电脑cpu核心数量.
     */
    @Test
    public void testMethod_availableProcessors() {
        System.out.println(Runtime.getRuntime().availableProcessors());
    }

    /**
     * 功能:
     *      测试传统方式计算的性能和stream方式的性能.
     * 输出:
     *      50000005000000
     *      stream产生10000000个数并求和的时间:249.6 ms
     *      50000005000000
     *      传统方式产生10000000个数并求和的时间:159.7 ms
     *      50000005000000
     *      并行流产生10000000个数并求和的时间:1.750 s
     *      50000005000000
     *      并行流取消拆箱装箱产生10000000个数并求和的时间:315.2 ms
     *      50000005000000
     *      使用LongStream产生10000000个数并求和的时间:21.98 ms
     *      50000005000000
     *      使用LongStream并行流产生10000000个数并求和的时间:15.94 ms
     * 说明:
     *      1. 传统方式和iterateStream简单流差异的时间不大.
     *          - 在计算大量数据时, 传统方式的执行时间更胜一筹;
     *      2. 由于iterateStream2方法中创建了并行流, 结果发现时间成本竟然上升了, 是因为stream中的元素为Long类型, 但是在计算时为long类型, 存在频繁的拆箱装箱操作, 于是:
     *          - 在iterateStream3方法中使用mapToLong将流转换为基本类型, 避免了拆箱装箱操作;
     *          - 在时间展示结果中发现, 一个流中如果存在大量的拆箱装箱操作, 则其效率可能反而会减低, 如果避免了拆箱装箱, 可以提升一定效率, 但是提升的幅度仍然不如使用传统方式计算的速度.
     *      3. 在iterateStream4方法中使用了LongStream这个Long类型的专用流来进行计算, 可以发现效率确实提升了;
     *          - 然后在iterateStream5方法时将这个流转换为并行流, 发现性能进一步提高, 提高的数量级到达了传统方式的十分之一.
     *      4. 所以, 使用并行操作需要恰当的操作, 不巧当的操作反而会导致性能的降低, 例如一个计算中存在大量的数据类型转换或者是拆箱装箱操作时, 就会使并行的执行效率低于传统方式实现的计算.
     *          - 这个提升的效率看中的就是cpu的能力, 就是拼cpu.
     *      5. java8 in action书中对并行处理的对象的能力分级:
     *          - ArrayList         Excellent(卓越的)
     *          - LinkedList        Poor(极差的)
     *          - IntStream.range   Excellent(卓越的)
     *          - Stream,iterate    Poor(极差的)
     *          - HashSet           Good(良好的)
     *          - TreeSet           Good(良好的)
     */
    @Test
    public void testMethod_() {
        Stopwatch started = Stopwatch.createStarted();
        System.out.println(iterateStream(10000000L));
        System.out.println("stream产生10000000个数并求和的时间:" + started.stop());

        started.reset().start();
        System.out.println(iterateStream_tradition(10000000L));
        System.out.println("传统方式产生10000000个数并求和的时间:" + started.stop());

        started.reset().start();
        System.out.println(iterateStream2(10000000L));
        System.out.println("并行流产生10000000个数并求和的时间:" + started.stop());

        started.reset().start();
        System.out.println(iterateStream3(10000000L));
        System.out.println("并行流取消拆箱装箱产生10000000个数并求和的时间:" + started.stop());

        started.reset().start();
        System.out.println(iterateStream4(10000000L));
        System.out.println("使用LongStream产生10000000个数并求和的时间:" + started.stop());

        started.reset().start();
        System.out.println(iterateStream5(10000000L));
        System.out.println("使用LongStream并行流产生10000000个数并求和的时间:" + started.stop());
    }

    private static Long iterateStream(Long limit) {
        Long reduce = Stream.iterate(1L, i -> i + 1).limit(limit).reduce(0L, Long::sum);
        return reduce;
    }

    private static Long iterateStream_tradition(Long limit) {
        Long result = 0L;
        for(Long i = 1L; i <= limit; i++) {
            result += i;
        }
        return result;
    }

    /**
     * 说明:
     *      1. 一个流使用parallel()方法会返回这个流的并行流, 就是使用多核cpu去运行.
     */
    private static Long iterateStream2(Long limit) {
        Long reduce = Stream.iterate(1L, i -> i + 1).parallel().limit(limit).reduce(0L, Long::sum);
        return reduce;
    }

    /**
     * 说明:
     *      1. mapToLong会将包装类型转换为基本类型, 从而避免计算过程中的拆箱装箱过程, 导致的浪费时间.
     */
    private static Long iterateStream3(Long limit) {
        long reduce = Stream.iterate(1L, i -> i + 1).mapToLong(Long::valueOf).parallel().limit(limit).reduce(0L, Long::sum);
        return reduce;
    }

    private static Long iterateStream4(Long limit) {
        long reduce = LongStream.rangeClosed(1L, limit).reduce(0L, Long::sum);
        return reduce;
    }

    private static Long iterateStream5(Long limit) {
        long reduce = LongStream.rangeClosed(1L, limit).parallel().reduce(0L, Long::sum);
        return reduce;
    }

}
